package integration.data.task.api.runner;

import integration.data.task.reader.Reader;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.record.RecordSender;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ReaderRunner extends AbstractRunner implements Runnable {

    private RecordSender recordSender;

    private Reader reader;

    public ReaderRunner(Reader reader) {
        this.reader = reader;
    }

    public void setRecordSender(RecordSender recordSender) {
        this.recordSender = recordSender;
    }

    public void setJobConf(String jobConf) {
        super.setJobConf(jobConf);
        reader.setConfig(jobConf);
    }

    @Override
    public void run() {
        assert null != this.recordSender;
        try {
            reader.init();
            reader.prepare();
            reader.startReader(recordSender);
            recordSender.terminate();
        } catch (Throwable e) {
            LogUtils.error(log, "Reader runner Received Exceptions:", e);
            super.markFail(e);
        } finally {
            log.debug("task reader starts to do destroy ...");
            this.destroy();
        }
    }

    @Override
    public void setTaskId(String taskId) {
        super.setTaskId(taskId);
        // reader taskID
        reader.setTaskId(taskId);
    }

    @Override
    public void setTaskPluginCollector(TaskPluginCollector pluginCollector) {
        reader.setTaskPluginCollector(pluginCollector);
    }

    public void shutdown() {
        recordSender.shutdown();
    }

    @Override
    public void destroy() {
        reader.destroy();
    }
}
